/*
 * sccilp.c, SmartCam coprocessor library for ILP processors
 *
 */

#define _USE_MATH_DEFINES
#define _XOPEN_SOURCE 500
#define _GNU_SOURCE

#include <string.h>
#include <errno.h>
#include <math.h>
#include <log.h>
#include <getopt.h>
#include <scos.h>
#include <glob.h>
#include <dlfcn.h>
#include <sched.h>

#include "sc.h"
#include "scc.h"
#include "scu.h"
#include "stat.h"
#include "scilp.h"

#include "settings.h"

#ifdef PSOS
#include <ops/custom_defs.h>
#else
#include "display.h"
#endif

log_default_streams_t ilpl;
static ilp_op_t  ilp_op[SC_MAX_OPERATIONS];
scu_buf_t ilp_buf[SC_MAX_BUFFERS];
static int running;
static int ilp_proc;
tmr_t displayTime, propTime;

extern unsigned char *disptr;

static glob_t seq_glob;
static int seq_nr=0;

scos_mutex_t ilp_mutex;
scos_cond_t ilp_cond;

int (*sched_setaffinity_len)(__pid_t __pid, unsigned int __len,
			     unsigned long int *__mask) __THROW = NULL;
int (*sched_setaffinity_nolen) (__pid_t __pid, __const unsigned long int *__mask) __THROW = NULL;

stat_t computeStat;

void scIlpInit(int proc)
{
  static int first=1;
  int i;
  scs_target_t *p;

  abortOnError(STREAM(ilpl.err), !first);

  logAddGroup();
  ilpl = logAddDefaultStreams(NULL);
  logEdit(ilpl.notice, LOG_GROUP_SET_LEVEL);

  lprintf(STREAM(ilpl.notice), "Initializing SmartCam ILP coprocessor %d", proc);
  ilp_proc = proc;

  for (i = 0; i < SC_MAX_BUFFERS; i++)
    ilp_buf[i].references = 0;

  for (i = 0; i < SC_MAX_OPERATIONS; i++)
    ilp_op[i].id = -1;

  glob(SCCILP_SEQUENCE_INPUT, 0, NULL, &seq_glob);

  scos_mutex_init(&ilp_mutex);
  scos_cond_init(&ilp_cond);

  lprintf(STREAM(ilpl.debug), "Capture from a sequence of %d elements", seq_glob.gl_pathc);

  sccTransportRegister(proc, ilpTransportReceiver, ilpBufferQuery);
  sccCoprocessorSetup(proc, ilpProcessMessage, sccTransportReceive);

#ifndef PSOS
  /* See if we can run the display operation */
  abortOnNULL(STREAM(ilpl.err), tblget(&scs_targets, proc, &p));
  if (tblget(&p->operations, DISPLAY, NULL))
    ilpDisplayInit(settings.argc, settings.argv);
#endif

  displayTime = tmrInit();
  computeStat = statInit();

  dlerror();
  sched_setaffinity_len = dlvsym(RTLD_DEFAULT, "sched_setaffinity", "GLIBC_2.3.4");
  if (!dlerror())
    lprintf(STREAM(ilpl.notice), "Using sched_setaffinity@GLIBC_2.3.4 to set affinity");
  else        
  {
    sched_setaffinity_len = NULL;
    sched_setaffinity_nolen = dlvsym(RTLD_DEFAULT, "sched_setaffinity", "GLIBC_2.3.3");
    if (!dlerror())
      lprintf(STREAM(ilpl.notice), "Using sched_setaffinity@GLIBC_2.3.3 to set affinity");
    else
    {
      sched_setaffinity_nolen = NULL;
      sched_setaffinity_len = dlsym(RTLD_DEFAULT, "sched_setaffinity");

      if (!dlerror())
        lprintf(STREAM(ilpl.notice), "Using sched_setaffinity to set affinity");
      else
        abortErrorl((STREAM(ilpl.err), "Couldn't find function to set affinity"));
    }
  }
}

void ilpProcessMessage(scc_hdr_t *hdr)
{
  scos_thread_t thread;
  ilp_op_t *op;
  scc_msg_t *msg = (scc_msg_t*)hdr;

  switch (msg->header.type)
  {
    case SCC_MSGTYPE_BUFFERCREATE:
      lprintf(STREAM(ilpl.debug), "Creating buffer %d (%p)", *((int*)msg->data),
        &ilp_buf[*((int*)msg->data)]);
      abortOnBounds(STREAM(ilpl.err), *((int*)msg->data), 0, SC_MAX_BUFFERS);

      if (ilp_buf[*((int*)msg->data)].references != 0)
	abortErrorl((STREAM(ilpl.err), "Buffer %d already exists", *((int*)msg->data)));

      ilp_buf[*((int*)msg->data)] = scuBufferCreate(msg->size,
					 	    SCU_BUF_MAX_READERS,
						    ilpBufferDestructor);
      break;
    case SCC_MSGTYPE_BUFFERFINALIZE:
      lprintf(STREAM(ilpl.debug), "Finalizing buffer %d", *((int*)msg->data));
      abortOnBounds(STREAM(ilpl.err), *((int*)msg->data), 0, SC_MAX_BUFFERS);
      scuBufferUnlock(&ilp_buf[*((int*)msg->data)]);
      break;
    case SCC_MSGTYPE_TRANSPORTCREATE:
      lprintf(STREAM(ilpl.debug), "Transmitting buffer %d to %d.%d",
	      *((int*)&msg->data[0]), msg->processor, *((int*)&msg->data[sizeof(int)]));
      sccTransportCreate(&ilp_buf[*((int*)&msg->data[0])], ilp_proc, 
			 msg->processor, *((int*)&msg->data[sizeof(int)]));
      break;
    case SCC_MSGTYPE_OPERATIONENQUEUE:
      lprintf(STREAM(ilpl.debug),
	      "Enqueueing operation %d type %d (needs %d args)",
	      msg->opid, msg->optype, msg->size);
      abortOnBounds(STREAM(ilpl.err), msg->opid, 0, SC_MAX_OPERATIONS);
      op = &ilp_op[msg->opid];

      if (op->id != -1)
	abortErrorl((STREAM(ilpl.err), "Operation %d already exists", msg->opid));

      op->id = msg->opid;
      op->operation = msg->optype;
      op->arguments = msg->size;      
      op->argreceived = 0;
      op->timer = tmrInit();
      break;
    case SCC_MSGTYPE_OPERATIONARGUMENT:
      abortOnBounds(STREAM(ilpl.err), msg->opid, 0, SC_MAX_OPERATIONS);
      abortOnBounds(STREAM(ilpl.err), msg->argid, 0, SC_MAX_ARGUMENTS);
      
      op = &ilp_op[msg->opid];
      op->argument[msg->argid].id = msg->argid;
      op->argument[msg->argid].type = msg->argtype;
      op->argument[msg->argid].intent = msg->argintent;

      switch (msg->argtype)
      {
	case SC_VAR_STREAM:
	  op->argument[msg->argid].data.d = *((int*)msg->data);
	  lprintf(STREAM(ilpl.debug),
		  "Received operation %d STREAM argument %d with value %d",	
          msg->opid, msg->argid, op->argument[msg->argid].data.d);

          op->argument[msg->argid].buffer =
            &ilp_buf[op->argument[msg->argid].data.d];

	  if (msg->argintent == SC_INTENT_IN)
	  {
	    /* To avoid adding a reader after the buffer has already been
	     * finalized (because the operation was waiting for another input),
	     * add the reader immediately.
	     */
	    
	    op->argument[msg->argid].reader =
	      scuBufferAddReader(op->argument[msg->argid].buffer);
	  }

	  break;
	case SC_VAR_INTEGER:
	  op->argument[msg->argid].data.d = *((int*)msg->data);
	  lprintf(STREAM(ilpl.debug),
		  "Received operation %d VAR_INTEGER argument %d with value %d",
		  msg->opid, msg->argid, op->argument[msg->argid].data.d);
	  break;
	case SC_VAR_DOUBLE:
	  op->argument[msg->argid].data.f = *((double*)msg->data);
	  lprintf(STREAM(ilpl.debug),
		  "Received operation %d VAR_DOUBLE argument %d with value %f",
		  msg->opid, msg->argid, op->argument[msg->argid].data.f);
	  break;
      }
      op->argreceived++;

      if (op->argreceived == op->arguments)
      {
      	lprintf(STREAM(ilpl.info), "Starting operation %d", op->id);

	running = 0;

	scos_mutex_lock(&ilp_mutex);

	scos_thread_create(&thread, ilpOperationEnqueue, (void*) op);

	scos_cond_wait(&ilp_cond, &ilp_mutex);
	scos_mutex_unlock(&ilp_mutex);

      	lprintf(STREAM(ilpl.info), "Done starting operation %d", op->id);
/*
        while (!running)
        {
          lprintf(STREAM(ilpl.debug), "Waiting for operation %d to start...\n", op->id);
          scos_yield();

#ifdef PSOS_
          if (!running)
          {
            abortErrorl((STREAM(ilpl.err), "Operation didn't start"));
          }
#endif
        }
*/
      }
      break;
   }
}

#ifdef PSOS
#define SCC_OPBUFFER    (640*10)
#define SCC_WIDTH       (640)
#define SCC_HEIGHT      (480)
#define SCC_OPBYTES     (SCC_WIDTH*SCC_HEIGHT)
#define SCC_IMWIDTH     (1280)
#else
#define SCC_IMWIDTH     (1024)
#define SCC_WIDTH       (1024)
#define SCC_HEIGHT      (1024)
#define SCC_OPBYTES     (SCC_WIDTH*SCC_HEIGHT)
#endif

void *ilpOperationEnqueue(void *operation)
{
  char namebuf[PATH_MAX];
  unsigned char buf[SCC_WIDTH];
  FILE *f;
  size_t bytes, lines;
  scu_bufptrid_t r;
  ilp_op_t *op = (ilp_op_t*)operation;
  unsigned long long affinity=(1<<settings.multiprocessing)-1;
  int result=0;
  long unsigned int seed = 1;

  scos_mutex_lock(&ilp_mutex);
  running = 1;
  scos_cond_signal(&ilp_cond);
  scos_mutex_unlock(&ilp_mutex);

  sprintf(namebuf, "op%d", op->id);
  logRegisterThread(namebuf);

  lprintf(STREAM(ilpl.info), "Started operation type %d", op->operation);

  /* Set affinity to the appropriate number of processors */
  if (sched_setaffinity_len)
    result = sched_setaffinity_len(0, sizeof(affinity), &affinity);
  else if (sched_setaffinity_nolen)
    result = sched_setaffinity_nolen(0, &affinity);
  else
    result = 1;

  if (result)
    abortErrorl((STREAM(ilpl.err), "Couldn't set affinity (%p %p) %llu for %d-processing (%d): %s (%d)",
            sched_setaffinity_len, sched_setaffinity_nolen,
            affinity, settings.multiprocessing, result, strerror(errno), errno));

  if (!memcmp(&computeStat, "\0\0\0\0", 4))
    statStart(&computeStat);

  tmrStart(&op->timer);

  switch (op->operation)
  {
    case READ:
      /* Read */
      sprintf(namebuf, SCCILP_IMAGE_INPUT, op->argument[0].data.d);
      lprintf(STREAM(ilpl.notice), "Reading, file='%s' (dI)", namebuf);
      abortOnNULL(STREAM(ilpl.err), f = fopen(namebuf, "r"));
      while ((bytes=fread(buf, 1, SCC_WIDTH, f)) > 0)
	scuBufferWrite(&ilp_buf[op->argument[1].data.d], buf, bytes,
		       SCU_BUF_BLOCK_ALL);
      scuBufferFinalize(&ilp_buf[op->argument[1].data.d]);
      fclose(f);
      break;
    case WRITE:
      /* Write */
      r = op->argument[0].reader;
      sprintf(namebuf, SCCILP_IMAGE_OUTPUT, op->argument[1].data.d);
      lprintf(STREAM(ilpl.notice), "Writing, file='%s' (id)", namebuf);
      abortOnNULL(STREAM(ilpl.err), f = fopen(namebuf, "w"));
      while (!scuBufferIsDry(&ilp_buf[op->argument[0].data.d], r))
      {
	bytes = scuBufferRead(&ilp_buf[op->argument[0].data.d], r,
			      buf, SCC_WIDTH, SCU_BUF_BLOCK_ANY);
	fwrite(buf, 1, bytes, f);
      }
      scuBufferRemoveReference(&ilp_buf[op->argument[0].data.d]);
      fclose(f);

      break;
    case GENERATE:
      /* Generate */
      lprintf(STREAM(ilpl.notice), "Generating (d=%d,I)",
	      op->argument[0].data.d);
	      
//      srand(0);
//      for (bytes=0; bytes < SCC_WIDTH; bytes++)
//        buf[bytes] = rand()&255;

      memset(buf, 0, SCC_WIDTH);

      for (lines=0; lines < op->argument[0].data.d; lines++)
      {
#if 1
	scuBufferAllocate(&ilp_buf[op->argument[1].data.d], (void**) &buf, SCC_WIDTH,
			  SCU_BUF_BLOCK_ALL);
	scuBufferReleaseAllocated(&ilp_buf[op->argument[1].data.d], SCC_WIDTH);
//	scuBufferWrite(&ilp_buf[op->argument[1].data.d], buf, SCC_WIDTH,
//		       SCU_BUF_BLOCK_ALL);
#else
	char *buf;
	int i;

	scuBufferAllocate(&ilp_buf[op->argument[1].data.d], (void**) &buf, SCC_WIDTH,
			  SCU_BUF_BLOCK_ALL);

	for (i=0; i < SCC_WIDTH; i++)
	{
          long unsigned int hi, lo;

	  buf[i] = seed;

#define constapmc 16807                                    

          /* http://www.firstpr.com.au/dsp/rand31/ */
          lo = constapmc * (seed & 0xFFFF);
          hi = constapmc * (seed >> 16);
          lo += (hi & 0x7FFF) << 16;
          lo += hi >> 15;                  
          if (lo > 0x7FFFFFFF) lo -= 0x7FFFFFFF;          
        
          seed = (long)lo;
	}
			
	scuBufferReleaseAllocated(&ilp_buf[op->argument[1].data.d], SCC_WIDTH);
#endif	
      }
      scuBufferFinalize(&ilp_buf[op->argument[1].data.d]);
      break;
    case DISCARD:
      /* Discard */
      r = op->argument[0].reader;
      lprintf(STREAM(ilpl.notice), "Discarding (i)");
      while (!scuBufferIsDry(&ilp_buf[op->argument[0].data.d], r))
      {
#if 0
	bytes = scuBufferRead(&ilp_buf[op->argument[0].data.d], r,
			      buf, SCC_WIDTH, SCU_BUF_BLOCK_ANY);
#else
	char *buf;

	bytes = scuBufferPeek(&ilp_buf[op->argument[0].data.d], r,
			      (void**) &buf, SCC_WIDTH, SCU_BUF_BLOCK_ANY);
	scuBufferReleasePeeked(&ilp_buf[op->argument[0].data.d], r, bytes);
#endif
      }
      scuBufferRemoveReference(&ilp_buf[op->argument[0].data.d]);

      break;
    case DISPLAY:
      {
        /* DISPLAY */
        int i = 0;
	r = op->argument[0].reader;

        lprintf(STREAM(ilpl.notice), "Displaying (i)");
        while (!scuBufferIsDry(&ilp_buf[op->argument[0].data.d], r))
        {
	  if (disptr)
	    scuBufferRead(&ilp_buf[op->argument[0].data.d], r, &disptr[i*SCC_IMWIDTH], SCC_WIDTH, SCU_BUF_BLOCK_ALL);
	  else
	    scuBufferRead(&ilp_buf[op->argument[0].data.d], r, buf, SCC_WIDTH, SCU_BUF_BLOCK_ALL);

          scos_trace(op->operation);
          if (!i) tmrStart(&displayTime);
          i++;
          if (i >= SCC_HEIGHT) i = SCC_HEIGHT-1;
        }
        tmrStop(&displayTime);
#ifndef PSOS
	ilpDisplayDamage();
#endif
        scuBufferRemoveReference(&ilp_buf[op->argument[0].data.d]);
      }
      break;
    case CAPTURE:
      /* Read */
      strcpy(namebuf, seq_glob.gl_pathv[seq_nr]);

      lprintf(STREAM(ilpl.notice), "Capturing, file='%s' (I)", namebuf);

      seq_nr++;
      if (seq_nr >= seq_glob.gl_pathc)
	seq_nr = 0;

      abortOnNULL(STREAM(ilpl.err), f = fopen(namebuf, "r"));
      while ((bytes=fread(buf, 1, SCC_WIDTH, f)) > 0)
	scuBufferWrite(&ilp_buf[op->argument[0].data.d], buf, bytes,
		       SCU_BUF_BLOCK_ALL);
      scuBufferFinalize(&ilp_buf[op->argument[0].data.d]);
      fclose(f);
      break;      
    case LOAD:
      /* Read */
      lprintf(STREAM(ilpl.notice), "Loading (d=%d, I)", op->argument[0].data.d);
      sprintf(namebuf, "%s/stream%d.scs", SCU_BUF_PATH, op->argument[0].data.d);
      abortOnNULL(STREAM(ilpl.err), f = fopen(namebuf, "r"));
      while ((bytes=fread(buf, 1, SCC_WIDTH, f)) > 0)
	scuBufferWrite(&ilp_buf[op->argument[1].data.d], buf, bytes,
		       SCU_BUF_BLOCK_ALL);
      scuBufferFinalize(&ilp_buf[op->argument[1].data.d]);
      fclose(f);
      break;
    case CYCLE:
      lprintf(STREAM(ilpl.notice), "Cycling (iI)");
      r = op->argument[0].reader;
      while (!scuBufferIsDry(&ilp_buf[op->argument[0].data.d], r))
      {
	bytes = scuBufferRead(&ilp_buf[op->argument[0].data.d], r,
			      buf, SCC_WIDTH, SCU_BUF_BLOCK_ANY);
	scuBufferWrite(&ilp_buf[op->argument[1].data.d], buf, bytes,
		       SCU_BUF_BLOCK_ALL);
      }
      scuBufferRemoveReference(&ilp_buf[op->argument[0].data.d]);
      scuBufferFinalize(&ilp_buf[op->argument[1].data.d]);
      break;
    case GENSIZE:
      lprintf(STREAM(ilpl.notice), "Generating from input size (iI)");
      r = op->argument[0].reader;
      while (!scuBufferIsDry(&ilp_buf[op->argument[0].data.d], r))
      {
        char *buf;
      
	bytes = scuBufferPeek(&ilp_buf[op->argument[0].data.d], r,
			      (void**)&buf, SCC_WIDTH, SCU_BUF_BLOCK_ANY);
	scuBufferReleasePeeked(&ilp_buf[op->argument[0].data.d], r, bytes);
	scuBufferAllocate(&ilp_buf[op->argument[1].data.d], (void**)&buf, bytes,
	                  SCU_BUF_BLOCK_ALL);
        scuBufferReleaseAllocated(&ilp_buf[op->argument[1].data.d], bytes);
      }
      scuBufferRemoveReference(&ilp_buf[op->argument[0].data.d]);
      scuBufferFinalize(&ilp_buf[op->argument[1].data.d]);
      break;
    case SPLIT:
      lprintf(STREAM(ilpl.notice), "Splitting (iII)");
      r = op->argument[0].reader;
      while (!scuBufferIsDry(&ilp_buf[op->argument[0].data.d], r))
      {
        char *buf;
      
	bytes = scuBufferPeek(&ilp_buf[op->argument[0].data.d], r,
			      (void**)&buf, SCC_WIDTH, SCU_BUF_BLOCK_ALL);
			      
        scuBufferWrite(&ilp_buf[op->argument[1].data.d], (void*)buf, bytes,
                       SCU_BUF_BLOCK_ALL);
        scuBufferWrite(&ilp_buf[op->argument[2].data.d], (void*)buf, bytes,
                       SCU_BUF_BLOCK_ALL);

	scuBufferReleasePeeked(&ilp_buf[op->argument[0].data.d], r, bytes);
      }
      scuBufferRemoveReference(&ilp_buf[op->argument[0].data.d]);
      scuBufferFinalize(&ilp_buf[op->argument[1].data.d]);
      scuBufferFinalize(&ilp_buf[op->argument[2].data.d]);
      break;
      
    default:
      ilpCustomOperationEnqueue(op);
      ilpMergedOperationEnqueue(op);
  }

  tmrStop(&op->timer);
  statStop(&computeStat);
  
  lprintf(STREAM(ilpl.info), "Finished at %.0f+%.0f = %.0f ms",
          statGetUserInUs(&computeStat)/1000.0, statGetSystemInUs(&computeStat)/1000.0,
          (statGetUserInUs(&computeStat)+statGetSystemInUs(&computeStat))/1000.0);

  ilpOperationComplete(op);

  return NULL;
}

ilp_op_t *ilpOperationAllocate(void)
{
  int i;

  for (i=0; i < SC_MAX_OPERATIONS; i++)
    if (ilp_op[i].id == -1)
    {
      lprintf(STREAM(ilpl.debug), "Allocating new operation structure %d", i);
      ilp_op[i].id = i;
      return &ilp_op[i];
    }

  lprintf(STREAM(ilpl.err), "Not enough room for new operation structure");  

  return NULL;  
}

ilp_op_t *ilpOperationFind(int id)
{
  int i;

  for (i=0; i < SC_MAX_OPERATIONS; i++)
    if (ilp_op[i].id == id)
      return &ilp_op[i];

  return NULL;
}

void ilpWriteResult(ilp_op_t *o, int id, int type, char *data)
{
  scc_msg_t msg;
 
  memset(&msg, 0, sizeof(scc_msg_t));
  msg.header.source = ilp_proc;
  msg.header.destination = SCC_MASTER_PROCESSOR;
  msg.header.size = sizeof(scc_msg_t)-sizeof(scc_hdr_t);
  msg.header.type = SCC_MSGTYPE_OPERATIONRESULT;

  msg.opid = o->id;
  msg.optype = o->operation;
  msg.argid = id;
  msg.argtype = type;
  
  switch (msg.argtype)
  {
    case SC_VAR_INTEGER:
      lprintf(STREAM(ilpl.debug),
	      "Sending operation %d VAR_INTEGER argument %d with value %d",
	      msg.opid, msg.argid, *((int*)data));
      memcpy(msg.data, data, sizeof(int));
      break;
    case SC_VAR_DOUBLE:
      lprintf(STREAM(ilpl.debug),
	      "Sending operation %d VAR_DOUBLE argument %d with value %f",
	      msg.opid, msg.argid, *((double*)data));
      memcpy(msg.data, data, sizeof(double));
      break;
  }

  sccSendMessage((scc_hdr_t*)&msg);
}

void ilpOperationComplete(ilp_op_t *o)
{
  scc_msg_t msg;
  
  memset(&msg, 0, sizeof(scc_msg_t));
  msg.header.source = ilp_proc;
  msg.header.destination = SCC_MASTER_PROCESSOR;
  msg.header.size = sizeof(scc_msg_t)-sizeof(scc_hdr_t);
  msg.header.type = SCC_MSGTYPE_OPERATIONCOMPLETED;

  msg.opid = o->id;
  msg.optype = o->operation;
  *((unsigned long long*)msg.data) = tmrGetValueInUs(&o->timer);

  /* Free */
  o->id = -1;
  
  sccSendMessage((scc_hdr_t*)&msg);
}

void ilpBufferDestructor(scu_buf_t *buf)
{
  scc_msg_t msg;

  memset(&msg, 0, sizeof(scc_msg_t));
  msg.header.source = ilp_proc;
  msg.header.destination = SCC_MASTER_PROCESSOR;
  msg.header.size = sizeof(scc_msg_t)-sizeof(scc_hdr_t);

  memcpy(&msg.data[0], &buf->total, sizeof(int));
  memcpy(&msg.data[sizeof(int)], &buf->counter, sizeof(int));

  if (buf->references)
  {
    int bufid = buf-ilp_buf;
    int i;
    
    lprintf(STREAM(ilpl.notice), "Writing deadlock on buffer %d", bufid);

    for (i=0; i < SC_MAX_OPERATIONS; i++)
    {
      ilp_op_t *op = &ilp_op[i];
    
      if (op->id != -1)
      {
        int j;
      
        for (j=0; j < op->arguments; j++)
        {
          if (op->argument[j].type == SC_VAR_STREAM && 
              op->argument[j].intent == SC_INTENT_OUT &&
              op->argument[j].data.d == bufid)
          {
            msg.opid = i;
            msg.optype = op->operation;
            msg.argid = j;
          }
        }
      }
    }
  
    lprintf(STREAM(ilpl.notice), "Operation type %d arg %d buffer must be increased",
            msg.optype, msg.argid);

    msg.header.type = SCC_MSGTYPE_BUFFERDEADLOCKED;
    msg.argtype = SC_VAR_STREAM;
    msg.argintent = SC_INTENT_OUT;
    msg.processor = ilp_proc;
    msg.size = buf->size;
  }
  else
  {
    lprintf(STREAM(ilpl.debug), "Buffer %d was destroyed", buf-ilp_buf);

    msg.header.type = SCC_MSGTYPE_BUFFERCOMPLETED;
    msg.argid = buf-ilp_buf;
    msg.argtype = SC_VAR_STREAM;
  }

  sccSendMessage((scc_hdr_t*)&msg);
}

unsigned long int ilpBufferQuery(int buffer)
{
  scu_buf_t *buf = &ilp_buf[buffer];

  if (!buf->references)
  {
    lprintf(STREAM(ilpl.warning), "Premature query for buffer %d", buf);
    return 0;
  }
  else
    return scuBufferSpaceAvailable(buf);
}

void ilpTransportReceiver(scc_hdr_t *hdr)
{
  scc_datamsg_t *msg = (scc_datamsg_t*)hdr;
  scu_buf_t *buf = &ilp_buf[msg->buffer], *db;

  if (!buf->references)
  {
    lprintf(STREAM(ilpl.warning), "Premature data for buffer %d, holding back", msg->buffer);
  
    while (!buf->references)
      scos_yield();
  }

  switch (hdr->type)
  {
    case SCC_MSGTYPE_BUFFERDATA:
      if (scuBufferSpaceAvailable(buf) < hdr->size-(sizeof(scc_datamsg_t)-sizeof(scc_hdr_t)))
        lprintf(STREAM(ilpl.crit), "Incoming communication of size %d for buffer %d (%d available) would block",
                hdr->size-(sizeof(scc_datamsg_t)-sizeof(scc_hdr_t)), msg->buffer, scuBufferSpaceAvailable(buf));

      scuBufferWrite(buf, msg->data, hdr->size-(sizeof(scc_datamsg_t)-sizeof(scc_hdr_t)),
		     SCU_BUF_BLOCK_ALL);
//        scuBufferAllocate(buf, (void*)&db, hdr->size-(sizeof(scc_datamsg_t)-sizeof(scc_hdr_t)),
//                           SCU_BUF_BLOCK_ALL);
//        scuBufferReleaseAllocated(buf, hdr->size-(sizeof(scc_datamsg_t)-sizeof(scc_hdr_t))); 
      break;
    case SCC_MSGTYPE_BUFFERDRY:
      scuBufferFinalize(buf);
      break;
    default:
      lprintf(STREAM(ilpl.err), "Unknown message type %d", hdr->type);
  }
}
